/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package fragment;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.table.api.TableEnvironment;
import utils.PasswordUtils;

/**
 * <pre>
 *     --MYSQL_HOST 127.0.0.1 --MYSQL_PORT 3306 --MYSQL_USER root --MYSQL_PASSWORD MTIzNDU2 --MYSQL_DATABASENAME xdcx --MYSQL_TABLESNAME xdcx.mk
 * </pre>
 */
public class MySQLSource implements SqlFragment {

    @Override
    public String createConnector(ParameterTool pt, TableEnvironment tableEnv) {
        String mysqlHost = pt.get("MYSQL_HOST");
        String mysqlPort = pt.get("MYSQL_PORT");
        String mysqlUser = pt.get("MYSQL_USER");
        String mysqlPassword = PasswordUtils.decode(pt.get("MYSQL_PASSWORD"));
        String mysqlDatabaseName = pt.get("MYSQL_DATABASENAME");
        String mysqlTablesName = pt.get("MYSQL_TABLESNAME");
        String startMode = pt.get("STARTMODE", "initial");
        String useSSL = pt.get("USESSL", "FALSE");

        StringBuilder optionalParameters = new StringBuilder();
        if (startMode.equalsIgnoreCase("specific-offset")) {
            String startupGtidSet = pt.get("SPECIFIC_OFFSET");
            optionalParameters.append(
                    String.format("'scan.startup.specific-offset.gtid-set' = '%s',\n", startupGtidSet));
        }

        String sourceSql = String.format(
                "CREATE TABLE `table_4827_mysql`(\n"
                        + "    `type` int,"
                        + "    `phase` int,"
                        + "    `data_source` string,"
                        + "    `db` string,"
                        + "    `schema` string,"
                        + "    `table` string,"
                        + "    `record` bytes,"
                        + "    `ddl` bytes,"
                        + "    `meta` string"
                        + ")\n"
                        + "    WITH (\n"
                        + "    'connector' = 'mysql-cdc-inlong',\n"
                        + "    'scan.newly-added-table.enabled' = 'true',\n"
                        + "    'include-schema-change' = 'true',\n"
                        + "    'include-incremental' = 'true',\n"
                        + "    'migrate-all' = 'true',\n"
                        + "    'scan.incremental.snapshot.enabled' = 'true',\n"
                        + "    'datasource.name' = 'db2',\n"
                        + "    'debezium.snapshot.locking.mode' = 'none',\n"
                        + "    'at-least-once-mode' = 'true',\n"
                        + "    'scan.lazy-calculate-splits.enabled' = 'false',\n"
                        + "    'job.version' = 'r4',\n"
                        + "    'server-time-zone' = 'Asia/Shanghai',\n"
                        + "    'inlong.table.metadata.columns' = 'opflag|meta.op_type,recordtime|meta.op_ts,receptime|meta.ts',\n"
                        + "    'oceanus.row-kind-string' = 'INS/INS/UPD/UPD',\n"
                        + "    'source.uid' = '4827_1',\n"

                        + "%s"

                        + "    'hostname' = '%s',\n"
                        + "    'port' = '%s',\n"
                        + "    'jdbc.properties.useSSL' = '%s',\n"
                        + "    'database-name' = '%s',\n"
                        + "    'table-name' = '%s',\n"
                        + "    'username' = '%s',\n"
                        + "    'password' = '%s',\n"
                        + "    'scan.startup.mode' = '%s'\n"
                        + ")", optionalParameters, mysqlHost, mysqlPort, useSSL, mysqlDatabaseName,
                mysqlTablesName, mysqlUser, mysqlPassword, startMode);
        LOGGER.info("source is \n {}", sourceSql);
        tableEnv.executeSql(sourceSql);

        return "table_4827_mysql";
    }
}
